package com.xzc.apitest.window

import com.xzc.apitest.source.SensorReading
import com.xzc.apitest.transform.MyFilter
import org.apache.flink.api.common.serialization.SimpleStringEncoder
import org.apache.flink.core.fs.Path
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{EventTimeSessionWindows, SlidingProcessingTimeWindows, TumblingEventTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time

object WindowTest {
  def main(args: Array[String]): Unit = {
    //创建执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //全局设置WM
    //    env.getConfig.setAutoWatermarkInterval(50)

    //2-文件
    //    val stream2 = env.readTextFile("D:\\git\\learning_flink\\_01_试用\\src\\main\\resources\\sensor.txt")
    //    stream2.print()

    val inputStream = env.socketTextStream("hadoop102", 7777);

    val dataStream = inputStream
      .map(data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })
      //理想的升序数据
      //      .assignAscendingTimestamps(_.timestamp * 1000L)
      //WM设置-有界无序，数据量大的情况就用这个，不是每来一个数据就生成一个WM
      //WM设置的大关窗慢-性能较好，设置小容易漏鱼-性能不是太好，按正态分布来说，可以设置小，放掉少量的鱼，通过侧输出流与滑窗来处理这些鱼
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(3)) {
        override def extractTimestamp(t: SensorReading): Long = t.timestamp * 1000L
      })

    val lateTag = new OutputTag[(String, Double, Long)]("late")

    val resultStream = dataStream
      .map(data => (data.id, data.temperature, data.timestamp))
      .keyBy(t => t._1)
      //滚动
      //      .window(TumblingEventTimeWindows.of(Time.seconds(15)))
      //滑动
      //      .window(SlidingProcessingTimeWindows.of(Time.seconds(15), Time.seconds(3)))
      //会话
      //      .window(EventTimeSessionWindows.withGap(Time.seconds(10)))
      //简写
      //      .countWindow(10) //滚动计算窗口
      .timeWindow(Time.seconds(15))
      //通过滑窗接住那些少量的鱼-这个延时1分钟也是针对的WM
      .allowedLateness(Time.minutes(1))
      //通过侧输出流继续接，如果有这个要求，保障数据不丢的情况下
      .sideOutputLateData(lateTag)
      //每15秒算一次，窗口内各传感器温度最小值
      //      .minBy(1)
      //以及最新的时间戳
      .reduce((curRes, newData) => (curRes._1, curRes._2.min(newData._2), newData._3))

    resultStream.getSideOutput(lateTag).print("late")
    resultStream.print("result")

    //执行
    env.execute("window test")
  }
}
